package io.rocketman.transport.netty.client;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.CharsetUtil;
import io.rocketman.common.thread.NamedThreadFactory;
import io.rocketman.transport.client.AbstractPeriodicityDetectionClient;
import io.rocketman.transport.client.OpenNetworkClientException;
import io.rocketman.transport.netty.channel.NettyChannel;
import io.rocketman.transport.netty.channel.NettyChannelConvertor;
import io.rocketman.transport.netty.client.handler.NettyClientChannelHandler;

import java.net.InetSocketAddress;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

/**
 * @author He Peng
 * @create 2018-05-25 2:22
 * @update 2018-05-25 2:22
 * @updatedesc : 更新说明
 * @see
 */
public class NettyClient extends AbstractPeriodicityDetectionClient {

    private Bootstrap bootstrap;
    private NettyChannel nettyChannel;
    private ChannelFuture connectFuture;

    private static final Executor EXECUTOR = Executors.newCachedThreadPool(new NamedThreadFactory("RocketmanNettyClientWorker"));
    private static final NettyChannelConvertor CHANNEL_CONVERTOR = new NettyChannelConvertor();
    private static final ByteBufAllocator BYTEBUF_ALLOCATOR = PooledByteBufAllocator.DEFAULT;

    public NettyClient(InetSocketAddress remoteAddress) throws OpenNetworkClientException {
        super(remoteAddress);
    }

    @Override
    protected void executeOpen(InetSocketAddress remoteAddress) throws Exception {
        NioEventLoopGroup group = new NioEventLoopGroup(0 , EXECUTOR);
        this.bootstrap = new Bootstrap();
        this.bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ch.pipeline().addLast(new NettyClientChannelHandler());
                    }
                });
        ChannelFuture channelFuture = this.bootstrap.connect(remoteAddress).sync();
        this.nettyChannel = CHANNEL_CONVERTOR.convert(channelFuture.channel());
        super.isOpen.compareAndSet(false , true);
    }

    @Override
    protected Future executeConnect() throws Exception {
        this.connectFuture = this.bootstrap.connect(this.remoteAddress);
        return this.connectFuture;
    }

    @Override
    protected void connectPostProcess() {
        boolean isCompleted = this.connectFuture.awaitUninterruptibly(1111, TimeUnit.MILLISECONDS);
        if (isCompleted && this.connectFuture.isSuccess()) {
            Channel newChannel = this.connectFuture.channel();
            if (this.nettyChannel != null && this.nettyChannel.isOpen()) {
                this.nettyChannel.close();
                this.nettyChannel = CHANNEL_CONVERTOR.convert(newChannel);
            }
        }
    }

    @Override
    protected Future getConnectFuture() {
        return this.connectFuture;
    }

    @Override
    public Future<Object> ping(Object message) {
        return null;
    }

    @Override
    protected Future executeRequest(Object message) throws Throwable {
        ByteBuf byteBuf = this.BYTEBUF_ALLOCATOR.buffer(1111111);
        byteBuf.writeBytes("TODO 编解码 、 序列化".getBytes(CharsetUtil.UTF_8));
        return this.nettyChannel.send(byteBuf);
    }

    @Override
    public io.rocketman.transport.channel.Channel getChannel() {
        return this.nettyChannel;
    }
}
